/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月19日 下午2:28:31
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月19日 下午2:28:31
*/
public class TransformerKeyBy {
	 public static void main(String[] args) throws Exception {
	        // 设置流执行环境
	        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	        // 加载数据源，并执行flatMap转换
	        DataStream<String> ds = env.fromElements("good good study","day day up")
	                .flatMap(new FlatMapFunction<String, String>() {
	                    /**
						 * 
						 */
						private static final long serialVersionUID = -7464308627786783866L;

						@Override
	                    public void flatMap(String value, Collector<String> out)
	                            throws Exception {
	                        for(String word: value.split("\\W+")){
	                            out.collect(word);
	                        }
	                    }
	                });
	        ds.print();
	        // 通过map转换，将事件流中事件的数据类型变换为(word,1)元组形式
	        DataStream<Tuple2<String,Integer>> ds_map =
	                ds.map(new MapFunction<String, Tuple2<String,Integer>>() {
	                    /**
						 * 
						 */
						private static final long serialVersionUID = 4651228081643315456L;

						@Override
	                    public Tuple2<String, Integer> map(String s) throws Exception {
	                        return new Tuple2<String,Integer>(s,1);
	                    }
	                });

	        // keyBy转换，按key重分区
	        KeyedStream<Tuple2<String,Integer>,Tuple> ds_keyed = ds_map.keyBy(0);

	        // 输出 
	        ds_keyed.print();

	        // 执行
	        env.execute("flink keyBy transformatiion");
	    }
}
